package com.dobbinsoft.iotechn.apigw.mq;

import com.alibaba.fastjson.JSONObject;
import com.dobbinsoft.iotechn.apigw.enums.WebsocketMessageType;
import com.dobbinsoft.iotechn.apigw.ws.DataPointWebsocket;
import com.dobbinsoft.iotechn.apigw.ws.message.DeviceDataPointWsMessage;
import com.dobbinsoft.iotechn.device.entity.DeviceDataSnapshotEntity;
import com.dobbinsoft.iotechn.device.mq.message.DeviceDataInnerMessage;
import com.dobbinsoft.iotechn.user.api.biz.UserBizService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.util.CollectionUtils;

import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Description: 数据点变化队列Listener
 * User: rize
 * Date: 2020/6/27
 * Time: 15:49
 */
public class DataPointInnerMessageListener implements MessageListener {

    @Autowired
    private Map dataWsSessionMap;

    @Autowired
    private UserBizService userBizService;

    @Override
    public void onMessage(Message message, byte[] topic) {
        // 接收到Redis的消息，从SessionMap中获取Session，并发送过去
        DeviceDataInnerMessage dataMessage = JSONObject.parseObject(message.toString(), DeviceDataInnerMessage.class);
        // 1.先要获取设备用户
        Long userId = dataMessage.getUserId();
        // 2.获取活跃用户（存在SESSION）
        Set<String> sessionSet = userBizService.getUserSessionSetById(userId);
        if (CollectionUtils.isEmpty(sessionSet)) {
            return;
        }
        // 3.获取活跃用户中连接上Socket的
        Set<String> wsSessionSet = dataWsSessionMap.keySet();
        Set<String> activeSessionSet = sessionSet.stream().filter(item -> wsSessionSet.contains(item)).collect(Collectors.toSet());
        if (CollectionUtils.isEmpty(activeSessionSet)) {
            return;
        }
        // 4.获取连接上Socket并且打开当前设备的
        for (String activeSession : activeSessionSet) {
            DataPointWebsocket dataPointWebsocket = (DataPointWebsocket) dataWsSessionMap.get(activeSession);
            Long currentDevice = dataPointWebsocket.getCurrentDevice();
            if (currentDevice != null && currentDevice.longValue() == dataMessage.getDeviceId().longValue()) {
                // 若是当前设备，则通知该设备数据变动
                DeviceDataPointWsMessage wsMessage = new DeviceDataPointWsMessage();
                wsMessage.setDeviceId(dataMessage.getDeviceId());
                DeviceDataSnapshotEntity entity = new DeviceDataSnapshotEntity();
                entity.setType(dataMessage.getType());
                entity.setIdTag(dataMessage.getIdTag());
                entity.setValue(dataMessage.getValue());
                wsMessage.setCode(WebsocketMessageType.DATA_CHANGE.getCode());
                wsMessage.setDataPoint(entity);
                dataPointWebsocket.sendMessage(JSONObject.toJSONString(wsMessage));
            }
        }
    }
}
